#![feature(async_closure)]

use std::collections::HashMap;
use std::convert::TryInto;
use std::error::Error;
use std::net::SocketAddr;
use std::sync::{Arc, Mutex};
use std::time::{SystemTime, UNIX_EPOCH};
use futures::FutureExt;
use bytes::{BufMut, Bytes, BytesMut};
use futures::SinkExt;
use futures::Sink;
use tracing_subscriber::layer::SubscriberExt;
use log::{error, info};
use tokio::net::UdpSocket;
use tokio_stream::StreamExt;
use tokio_util::udp::UdpFramed;
use crate::codec::d_codec::{Packet, PacketBytesCodec, Transmission};
use crate::config::Config;
use crate::serializer::serialization::{DeSerialized, Serialized};
use crate::start::Connection;
use crate::start::server::ServiceBuilder;
use tokio::time;
use tracing_subscriber::fmt;
use tracing_subscriber::util::SubscriberInitExt;
use crate::start::dispatcher::Handler;

type PeerId = u128;

#[derive(Debug)]
pub struct Node {
    peer_id: PeerId,
    address: SocketAddr,
    create_time: u64,
    keep_live_time: u64,
}


pub struct NodeManager {
    current_node: Node,
    inner: Arc<Mutex<NodeManagerInner>>,
    // inner:Mutex<Arc<NodeManagerInner>>
}

pub struct NodeManagerInner {
    nodes: HashMap<PeerId, Node>,
}

impl NodeManagerInner {
    fn node_heart(mut self, node: Node) {
        self.nodes.entry(node.peer_id).or_insert(Node {
            peer_id: node.peer_id,
            address: node.address,
            create_time: node.create_time,
            keep_live_time: 0,
        });
        info!("recv: {:?}", node);
    }
}

impl NodeManager {
    pub fn init() -> Self {
        let peer_id = uuid::Uuid::new_v4().as_u128();
        NodeManager {
            current_node: Node {
                peer_id: peer_id,
                address: SocketAddr::new("127.0.0.1".parse().unwrap(), 0),
                create_time: SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs(),
                keep_live_time: 0,
            },
            inner: Arc::new(Mutex::new(NodeManagerInner {
                nodes: HashMap::new(),
            })),
        }
    }
    pub async fn start<F: 'static + Fn(Connection) + Sync + Send>(node_manager: Arc<NodeManager>, config: Config, handler: Option<HashMap<i16, F>>) -> Result<(), Box<dyn Error>> {
        let node = &node_manager.current_node;
        let info = NodeInfo {
            peer_id: node.peer_id,
            create_time: node.create_time,
        };
        let config_clone = config.clone();
        tokio::spawn(async move {
            Self::broadcast_node(config.clone(), info).await;
        });
        let node_manager_clone = node_manager.clone();
        tokio::spawn(async move {
            Self::clear_timeout_node(&node_manager_clone).await;
        });
        let mut builder = ServiceBuilder::default(config_clone.clone());
        if let Some(handler) = handler {
            for f in handler {
                builder = builder.register(f.0, f.1);
            }
        }
        if config_clone.node_heart {
            builder = builder.register(i16::MIN, move |connection: Connection| {
                let mut node_info = NodeInfo::default();
                connection.de_serialized(&mut node_info);
                info!("receive: {:?}", node_info);
                let mut inner = node_manager.inner.lock().unwrap();
                inner.nodes.entry(node_info.peer_id).or_insert(Node {
                    peer_id: node_info.peer_id,
                    address: *connection.get_socket_addr(),
                    create_time: node_info.create_time,
                    // todo 这个时间可以用来记录每个节点的心跳时间，防止同一时间一起ping
                    keep_live_time: SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs(),
                });
            });
        }
        builder
            .builder()
            .start().await;
        return Ok(());
    }

    async fn clear_timeout_node(node_manager: &Arc<NodeManager>) {
        let mut interval = time::interval(time::Duration::from_secs(2));
        loop {
            interval.tick().await;
            let mut inner = node_manager.inner.lock().unwrap();
            let now = SystemTime::now();

            let timeout_duration = std::time::Duration::from_secs(3);
            inner.nodes.retain(|_, node| {
                let system_time = UNIX_EPOCH + std::time::Duration::from_secs(node.keep_live_time);
                let duration = now.duration_since(system_time).expect("Time went backwards");
                duration <= timeout_duration
            });
        }
    }

    async fn broadcast_node(config: Config, info: NodeInfo) {
        let mut interval = time::interval(time::Duration::from_secs(2));

        //   定时任务  广播self
        let local_addr: SocketAddr = if config.address.is_ipv4() {
            "0.0.0.0:0"
        } else {
            "[::]:0"
        }
            .parse().unwrap();
        let result = UdpSocket::bind(local_addr).await;
        let socket = result.unwrap();
        socket.set_broadcast(true).unwrap();
        // 格式化地址
        let addr = format!("{}:{}", "255.255.255.255", config.port);
        let mut framed = UdpFramed::new(socket, PacketBytesCodec::default());
        let mut packet = Packet::default();


        let bytes = info.serialized();
        packet.set_cmd(i16::MIN);
        packet.set_body_length(bytes.len() as i32);
        packet.set_body(bytes);
        loop {
            let result = framed.send((packet.clone(), addr.parse().unwrap())).await;
            if result.is_err() {
                error!("send error: {:?}", result.err().unwrap());
            }
            info!("send: {:?}", packet);
            interval.tick().await;
        }
    }
}

#[derive(Default, Debug)]
pub struct NodeInfo {
    peer_id: u128,
    create_time: u64,
}

impl Serialized for NodeInfo {
    fn serialized(&self) -> Bytes {
        let mut result = BytesMut::new();

        result.put_u128(self.peer_id);
        result.put_u64(self.create_time);

        result.freeze()
    }
}

impl DeSerialized for NodeInfo {
    // type Obj = NodeInfo;

    fn deserialized(&mut self, body: &Bytes) -> &mut Self {
        // Bytes::from(body.to_vec())
        let data = body.to_vec();
        let peer_id = u128::from_be_bytes(data[0..16].try_into().unwrap());
        let create_time = u64::from_be_bytes(data[16..24].try_into().unwrap());
        // NodeInfo { peer_id, create_time }
        self.peer_id = peer_id;
        self.create_time = create_time;
        self
    }
}

impl Node {}

pub struct NodePool {
    node_map: HashMap<u32, Node>,
}

#[tokio::test]
async fn test_udp_broadcast() {
    let result = UdpSocket::bind("0.0.0.0:12345").await;
    let socket = result.unwrap();

    const MAX_DATAGRAM_SIZE: usize = 65_507;
    let mut data = vec![0u8; MAX_DATAGRAM_SIZE];
    let mut framed = UdpFramed::new(socket, PacketBytesCodec::default());
    let (bytes, addr) = framed.next().map(|e| e.unwrap()).await.unwrap();

    println!("[a] recv: {:?}", bytes);
    // let len = socket.recv(&mut data).await.unwrap();
    // println!(
    //     "Received {} bytes:\n{}",
    //     len,
    //     String::from_utf8_lossy(&data[..len])
    // );
}

#[tokio::test]
async fn test_udp() {
    let result = UdpSocket::bind("0.0.0.0:0").await;
    let socket = result.unwrap();
    socket.set_broadcast(true).unwrap();

    let mut framed = UdpFramed::new(socket, PacketBytesCodec::default());
    let mut packet = Packet::default();

    let info = NodeInfo {
        peer_id: 1231232,
        create_time: 1232,
    };
    let bytes = info.serialized();
    packet.set_body_length(bytes.len() as i32);
    packet.set_body(bytes);
    framed.send((packet, "255.255.255.255:12345".parse().unwrap())).await.unwrap();
    // let b = b"hello";
    // socket.send_to(b, "255.255.255.255:12345").await.unwrap();
}

// #[cfg(test)]
// fn setup() {
//     let _ = env_logger::builder().is_test(true).try_init();
// }
#[tokio::test]
// #[cfg(test)]
async fn test_node_manager() {
    tracing_subscriber::registry()
        .with(fmt::layer())
        .init();
    info!("test_node_manager");
    let manager = NodeManager::init();
    let mut handler = HashMap::new();
    handler.insert(1, |connection: Connection| {
        let mut node_info = NodeInfo::default();
        connection.de_serialized(&mut node_info);
        println!("recv: {:?}", node_info);
    });
    let mut config = Config::debug_default_udp();
    // config.port=8090;
    NodeManager::start(Arc::new(manager), config, Some(handler)).await.expect("start error");
}